package nats

import (
	"lnzw/engine/log"
	"sync"
	"time"

	"github.com/nats-io/nats.go"
	"go.uber.org/zap"
)

var (
	once       sync.Once
	GlobalNats *nats.Conn
)

func Setup() {
	once.Do(func() {
		log.Info("start nats connecting...")
		var err error
		GlobalNats, err = nats.Connect("nats://192.168.31.70:24222",
			nats.MaxReconnects(10),                        // 重连次数
			nats.ReconnectWait(nats.DefaultReconnectWait), // 重连间隔
			nats.ReconnectHandler(func(nc *nats.Conn) {
				log.Info("nats.Setup", zap.String("reconnect", nc.ConnectedUrl()))
			}), // 重连回调
			nats.ConnectHandler(func(nc *nats.Conn) {
				log.Info("nats.Setup", zap.String("connect", nc.ConnectedUrl()))
			}), // 连接成功回调
			nats.DisconnectHandler(func(nc *nats.Conn) {
				log.Info("nats.Setup", zap.String("disconnect", nc.ConnectedUrl()))
			}), // 断开连接回调
			nats.Name("nats-client"),          // 客户端名称
			nats.Timeout(nats.DefaultTimeout), // 连接超时
		)
		if err != nil {
			log.Error("nats.Setup", zap.Error(err))
			panic(err)
		}
	})
	time.Sleep(3 * time.Second)
}

// 同步reply (使用一对多的)
func SyncQueueReply(subject string, queue string, cb func(msg []byte, ch chan<- []byte)) error {

	sub, err := GlobalNats.QueueSubscribeSync(subject, "test")
	if err != nil {
		log.Error("nats.SyncReply", zap.Error(err))
		return err
	}
	for {
		msg, err := sub.NextMsg(time.Hour)
		if err != nil {
			log.Error("nats.SyncReply", zap.Error(err))
			return err
		}
		ch := make(chan []byte)
		defer close(ch)
		go cb(msg.Data, ch)
		select {
		case data := <-ch:
			msg.Respond(data)
		case <-time.After(time.Second):
			log.Warn("nats.AsyncReply reply timeout")
		}
	}
}

// 同步reply (使用一对多的)
func SyncReply(subject string, cb func(msg []byte, ch chan<- []byte)) error {

	sub, err := GlobalNats.SubscribeSync(subject)
	if err != nil {
		log.Error("nats.SyncReply", zap.Error(err))
		return err
	}
	for {
		msg, err := sub.NextMsg(time.Hour)
		if err != nil {
			log.Error("nats.SyncReply", zap.Error(err))
			return err
		}
		ch := make(chan []byte)
		defer close(ch)
		go cb(msg.Data, ch)
		select {
		case data := <-ch:
			msg.Respond(data)
		case <-time.After(time.Second):
			log.Warn("nats.AsyncReply reply timeout")
		}
	}
}

// 异步reply
func AsyncReply(subject string, cb func(msg []byte, ch chan<- []byte)) {
	GlobalNats.Subscribe(subject, func(msg *nats.Msg) {
		ch := make(chan []byte)
		defer close(ch)
		go cb(msg.Data, ch)
		select {
		case data := <-ch:
			msg.Respond(data)
		case <-time.After(time.Second):
			log.Warn("nats.AsyncReply reply timeout")
		}
	})
}

// 异步订阅
func AsynSubscribe(subject string, cb func(msg []byte)) error {
	_, err := GlobalNats.Subscribe(subject, func(msg *nats.Msg) {
		cb(msg.Data)
	})
	if err != nil {
		log.Error("nats.Subscribe", zap.Error(err))
	}
	return err
}

// 同步订阅
func SyncSubscribe(subject string, cb func(bs []byte)) error {
	sub, err := GlobalNats.SubscribeSync(subject)
	if err != nil {
		log.Error("nats.SubscribeSync", zap.Error(err))
		return err
	}
	for {
		msg, err := sub.NextMsg(time.Hour)
		if err != nil {
			log.Error("nats.SubscribeSync", zap.Error(err))
			continue
		}
		cb(msg.Data)
	}
}

// 发布
func Publish(subject string, data []byte, sync bool) error {
	err := GlobalNats.Publish(subject, data)
	if err != nil {
		log.Error("nats.Publish", zap.Error(err))
		return err
	}
	if sync {
		err = GlobalNats.Flush()
		if err != nil {
			log.Error("nats.Flush", zap.Error(err))
			return err
		}
	}
	return nil
}

func Request(subject string, data []byte) ([]byte, error) {
	msg := &nats.Msg{
		Subject: subject,
		Data:    data,
		Reply:   subject + ".reply",
	}

	rep, err := GlobalNats.RequestMsg(msg, time.Second)
	if err != nil {
		log.Error("nats.Request", zap.Error(err))
		return nil, err
	}
	return rep.Data, nil
}

// 同步队列组
func SyncQueueSubscribe(subject string, queue string, cb func(bs []byte)) error {
	sub, err := GlobalNats.QueueSubscribeSync(subject, queue)
	if err != nil {
		log.Error("nats.QueueSubscribe", zap.Error(err))
		return err
	}
	for {
		msg, err := sub.NextMsg(time.Hour)
		if err != nil {
			log.Error("nats.QueueSubscribe", zap.Error(err))
			continue
		}
		cb(msg.Data)
	}
}

// 异步队列组
func AsynQueueSubscribe(subject string, queue string, cb func(bs []byte)) error {
	_, err := GlobalNats.QueueSubscribe(subject, queue, func(msg *nats.Msg) {
		cb(msg.Data)
	})
	if err != nil {
		log.Error("nats.QueueSubscribe", zap.Error(err))
	}
	return err
}

func Close() {
	if GlobalNats != nil {
		GlobalNats.Close()
	}
}
